/*
* This file is a part of the open-eBackup project.
* This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
* If a copy of the MPL was not distributed with this file, You can obtain one at
* http://mozilla.org/MPL/2.0/.
*
* Copyright (c) [2024] Huawei Technologies Co.,Ltd.
*
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
*/
#include <boost/filesystem.hpp>
#include "common/EnvVarManager.h"
#include "common/Utils.h"
#include "utils/PluginUtilities.h"
#include "system/System.hpp"
#include "client/ClientInvoke.h"
#include "log/BackupFailureRecorder.h"
#include "ObjectStorageCommonService.h"

using namespace std;
using namespace Module;

namespace OBSPlugin {
namespace {
    constexpr auto MODULE = "ObjectStorageCommonService";
    constexpr uint32_t NUMBER0 = 0;
    constexpr uint32_t NUMBER3 = 3;
    constexpr uint32_t NUMBER4  = 4;
    constexpr uint32_t NUMBER5  = 5;
    constexpr uint32_t NUMBER10 = 10;
    constexpr uint32_t NUMBER50  = 50;
    constexpr uint32_t NUMBER60  = 60;
    constexpr uint32_t NUMBER100  = 100;
    constexpr uint32_t NUMBER256 = 256;
    // X3000 Internal Agent Max Channel number
    const int INTERNAL_MAX_CHANNEL_NUM = 15;
    const std::string PLUGIN_ATT_JSON = "plugin_attribute_1.0.0.json";
    const std::string DEFAULT_FAILURE_DEFAULT_ROOT =
        R"(\opt\DataBackup\ProtectClient\ProtectClient-E\log\Plugins\ObsPlugin)";
    const uint64_t DEFAULT_MAX_FAILURE_RECORDS_NUM = 1000000;
    constexpr uint64_t RETRY_ERR_CODE = 1593987332;
    constexpr uint32_t SCANNER_REPORT_CIRCLE_TIME = 60;  /* seconds */
    const std::string OBS_CONFIG_FILE = "/opt/protectengine-OBS-conf/";
}

NodeLevelTaskInfo ObjectStorageCommonService::m_nodeLevelTaskInfo {};

ObjectStorageCommonService::ObjectStorageCommonService() : m_scanMgr(ScanMgr::GetInstance())
{
    /* use logger root as the failure records output root */
    m_failureRecordRoot = Module::CLogger::GetInstance().GetLogRootPath();
    INFOLOG("using logger root path %s as default root path for failure recorder", m_failureRecordRoot.c_str());
    if (m_failureRecordRoot.empty()) {
        WARNLOG("got empty failure record root from log root, set to default");
        m_failureRecordRoot = DEFAULT_FAILURE_DEFAULT_ROOT;
    }
    m_maxFailureRecordsNum = DEFAULT_MAX_FAILURE_RECORDS_NUM;
}

bool ObjectStorageCommonService::InitIdGenerator()
{
    m_idGenerator = std::make_shared<Module::Snowflake>();
    if (m_idGenerator == nullptr) {
        ERRLOG("Init idGenerator failed, iter is nullptr. jobid: %s", m_jobId.c_str());
        return false;
    }
    size_t machineId = Module::GetMachineId();
    m_idGenerator->SetMachine(machineId);
    return true;
}

bool ObjectStorageCommonService::checkFilePathAndGetSrcFileList(std::string srcDir, std::string dstDir,
    std::vector<std::string> &srcFileList)
{
    if (!IsPathExists(srcDir)) {
        HCP_Log(ERR, MODULE) << "SrcDir does not exist: " << srcDir << HCPENDLOG;
        return false;
    }

    if (!IsPathExists(dstDir)) {
        HCP_Log(ERR, MODULE) << "DstDir does not exist: " << dstDir << HCPENDLOG;
        return false;
    }

    if (!GetFileListInDirectory(srcDir, srcFileList)) {
        HCP_Log(ERR, MODULE) << "Get filelist for dir failed: " << srcDir << HCPENDLOG;
        return false;
    }

    return true;
}

bool ObjectStorageCommonService::IsValidCtrlFile(uint32_t subTaskType, std::string ctrlFileFullPath)
{
    /* "txt.tmp" are temproary files and is considered as not valid for all the phases */
    if (ctrlFileFullPath.find("txt.tmp") != std::string::npos) {
        return false;
    }

    switch (subTaskType) {
        case OBSPlugin::SUBJOB_TYPE_DATACOPY_COPY_PHASE: {
            return ctrlFileFullPath.find("/control_") != std::string::npos;
        }
        case OBSPlugin::SUBJOB_TYPE_DATACOPY_DELETE_PHASE: {
            return ctrlFileFullPath.find("/delete_control_") != std::string::npos;
        }
        case SUBJOB_TYPE_ANY: {
            return (ctrlFileFullPath.find("/control_") != std::string::npos) ||
                (ctrlFileFullPath.find("/delete_control_") != std::string::npos);
        }
        default: {
            return false;
        }
    }
}

bool ObjectStorageCommonService::GetSubJobTypeByFileName(const std::string &fileName, std::string &subTaskName,
    uint32_t &subTaskType, uint32_t &subTaskPrio)
{
    if (m_idGenerator == nullptr) {
        InitIdGenerator();
    }
    if (fileName.find("delete") != std::string::npos) {
        subTaskName = "ObsBackup_DelFile_" + std::to_string(m_idGenerator->GenerateId());
        subTaskType = OBSPlugin::SUBJOB_TYPE_DATACOPY_DELETE_PHASE;
        subTaskPrio = OBSPlugin::SUBJOB_TYPE_DATACOPY_DELETE_PHASE_PRIO;
    } else if (fileName.find("/control_") != std::string::npos) {
        subTaskName = "ObsBackup_CtrlFile_" + std::to_string(m_idGenerator->GenerateId());
        subTaskType = OBSPlugin::SUBJOB_TYPE_DATACOPY_COPY_PHASE;
        subTaskPrio = OBSPlugin::SUBJOB_TYPE_DATACOPY_COPY_PHASE_PRIO;
    }

    return true;
}

bool ObjectStorageCommonService::InitSubTask(SubJob &subJob, const std::string& jobId,
    const std::string& ctrlFile, uint32_t &subTaskType, const std::string& ext)
{
    std::string subTaskName {};
    uint32_t subTaskPrio {};
    std::string backupSubJobStr {};
    ObjectStorageBackupSubJob backupSubJob {};
    GetSubJobTypeByFileName(ctrlFile, subTaskName, subTaskType, subTaskPrio);
    backupSubJob.m_ControlFile = ctrlFile;
    backupSubJob.m_SubTaskType = subTaskType;
    backupSubJob.m_Ext = ext;
    if (!Module::JsonHelper::StructToJsonString(backupSubJob, backupSubJobStr)) {
        HCP_Log(ERR, MODULE) << "Convert to json failed for file: "
        << WIPE_SENSITIVE(backupSubJob.m_ControlFile) << HCPENDLOG;
        return false;
    }
    subJob.__set_jobId(jobId);
    subJob.__set_jobName(subTaskName);
    subJob.__set_jobType(SubJobType::BUSINESS_SUB_JOB);
    subJob.__set_policy(ExecutePolicy::ANY_NODE);
    subJob.__set_jobInfo(backupSubJobStr);
    subJob.__set_jobPriority(subTaskPrio);
    subJob.__set_ignoreFailed(false);
    return true;
}

int ObjectStorageCommonService::CreateSubTask(std::vector<SubJob> &subJobList,
    std::vector<std::string> &ctrlFileList, bool isCheckPoint)
{
    if (subJobList.empty() || ctrlFileList.empty()) {
        DBGLOG("subJobList or ctrlFileList are empty, exit CreateSubTask");
        return Module::SUCCESS;
    }
    INFOLOG("Enter CreateSubTask, jobId: %s subJobList size = %d", m_jobId.c_str(), subJobList.size());
    for (const SubJob& subJob: subJobList) {
        HCP_Log(INFO, MODULE) << "Create subtask, jobId: " << subJob.jobId << ", jobName: " << subJob.jobName
            << ", jobType: " << subJob.jobType << ", jobPrio: " << subJob.jobPriority
            << ", jobInfo: " << subJob.jobInfo << ", time: " << OBSPlugin::GetCurrentTimeInSeconds() << HCPENDLOG;
    }
    ActionResult ret;
    int retryCnt = 0;
    do {
        if (retryCnt == OBSPlugin::MIDDLE_NORMAL_RETRY_TIMES) {
            ObjectStorageCommonService::KeepJobAlive(m_jobId, m_subJobId);
        }
        if (IsAbortJob()) {
            HCP_Log(WARN, MODULE) << "Abort Recvied taskid: " << m_jobId << ", subtaskid: " << m_subJobId << HCPENDLOG;
            break;
        }
        InterfaceInvokeTimer(
            std::chrono::seconds(NUMBER10),
            [&](...) { JobService::AddNewJob(ret, subJobList); },
            [&](...) { WARNLOG("JobService::AddNewJob timeout!"); });
        if (ret.code == Module::SUCCESS) {
            DBGLOG("Exit AddNewJob, success, jobId: %s", m_jobId.c_str());
            break;
        } else if ((ret.bodyErr == RETRY_ERR_CODE) && (ret.code == Module::INTERNAL_ERR)) {
            WARNLOG("Exit AddNewJob, retry, jobId: %s retCode: %u", m_jobId.c_str(), ret.code);
            return Module::RETRY;
        }
        Module::SleepFor(std::chrono::seconds(OBSPlugin::NORMAL_RETRY_INTERVAL));
        WARNLOG("AddNewJob for main task: %s failed: retCode: %u", m_jobId.c_str(), ret.code);
    } while (++retryCnt <= OBSPlugin::NORMAL_RETRY_TIMES);
    if (ret.code != MP_SUCCESS) {
        ERRLOG("Exit AddNewJob, failed after max retry, ret.code: %d, jobId: %s", ret.code, m_jobId.c_str());
        return Module::FAILED;
    }
    for (const std::string& ctrlFile : ctrlFileList) {
        if (!isCheckPoint) {
            RemoveFile(ctrlFile);
        }
        m_nodeLevelTaskInfo.IncrSubTasksSubmitCount(); // Increment the task submit count
    }
    subJobList.clear();
    ctrlFileList.clear();
    DBGLOG("Exit CreateSubTasksFromCtrlFile, CreateSubTask");
    return Module::SUCCESS;
}

int ObjectStorageCommonService::CreateSubTask(const SubJob &subJob)
{
    HCP_Log(INFO, MODULE)
        << "Create subtask, jobId: " << subJob.jobId
        << ", jobName: " << subJob.jobName
        << ", jobType: " << subJob.jobType
        << ", jobPrio: " << subJob.jobPriority
        << ", jobInfo: " << subJob.jobInfo
        // << ", time: " << GetCurrentTimeInSeconds()
        << HCPENDLOG;
    std::vector<SubJob> subJobList;
    subJobList.emplace_back(subJob);
    ActionResult ret;
    int retryCnt = 0;
    do {
        if (IsAbortJob()) {
            INFOLOG("Exit received Abort for taskid: %s, subtaskid: %s", m_jobId.c_str(), m_subJobId.c_str());
            return Module::SUCCESS;
        }
        JobService::AddNewJob(ret, subJobList);
        if (ret.code == Module::SUCCESS) {
            DBGLOG("Create SubTask success, jobId: %s", m_jobId.c_str());
            return Module::SUCCESS;
        } else if ((ret.bodyErr == RETRY_ERR_CODE) && (ret.code == Module::INTERNAL_ERR)) {
            DBGLOG("Exit AddNewJob, retry %d, jobId: %s", retryCnt, m_jobId.c_str());
            return Module::RETRY;
        }
        Module::SleepFor(std::chrono::seconds(OBSPlugin::NORMAL_RETRY_INTERVAL));
        WARNLOG("Create SubTask failed for main task: %s failed: %u bodyErr: %u",
            m_jobId.c_str(), ret.code, ret.bodyErr);
    } while (++retryCnt <= OBSPlugin::NORMAL_RETRY_TIMES);
    ERRLOG("Create SubTask failed after max retry, ret.code:%d, jobId: %s", ret.code, m_jobId.c_str());
    return Module::FAILED;
}

bool ObjectStorageCommonService::ReportJobDetails(const SubJobDetails &jobInfo)
{
    ActionResult ret;
    HCP_Log(INFO, MODULE) << "ReportJobDetails "
        << ", jobId: " << jobInfo.jobId
        << ", subJobId: " << jobInfo.subJobId
        << ", jobStatus: " << jobInfo.jobStatus
        << ", progress: " << jobInfo.progress
        << ", dataSize: " << jobInfo.dataSize
        << ", level: " << jobInfo.logDetail[0].level
        << ", description: " << jobInfo.logDetail[0].description << HCPENDLOG;

    HCP_Log(INFO, MODULE) << "Start JobService::ReportJobDetails" <<  HCPENDLOG;
    JobService::ReportJobDetails(ret,  jobInfo);
    HCP_Log(INFO, MODULE) << "Completed JobService::ReportJobDetails" <<  HCPENDLOG;
    if (ret.code != MP_SUCCESS) {
        HCP_Log(ERR, MODULE) << "Report job details to controller failed: " << ret.code <<  HCPENDLOG;
        return false;
    }
    return true;
}

bool ObjectStorageCommonService::CreateResource(const std::string key, const std::string &val,
    const std::string& mainJobId, const std::string& subJobId)
{
    int retryCnt = 0;
    ActionResult ret {};
    Resource res {};
    res.__set_sharedNum(1);
    res.__set_scope(ResourceScope::SINGLE_JOB);
    res.__set_scopeKey(key);
    res.__set_resourceKey(key);
    res.__set_resourceValue(val);

    do {
        if (retryCnt == OBSPlugin::MIDDLE_NORMAL_RETRY_TIMES) {
            ObjectStorageCommonService::KeepJobAlive(mainJobId, subJobId);
        }
        ShareResource::CreateResource(ret, res, mainJobId);
        if (ret.code != Module::FAILED) {
            DBGLOG("Create resource success for key: %s!", key.c_str());
            return true;
        }
        Module::SleepFor(std::chrono::seconds(OBSPlugin::NORMAL_RETRY_INTERVAL));
        WARNLOG("CreateResource for key: %s failed, retryCnt: %d", key.c_str(), retryCnt);
    } while (++retryCnt <= NORMAL_RETRY_TIMES);
    ERRLOG("Create resource failed for key: %s, ret.code: %d", key.c_str(), ret.code);
    return false;
}

bool ObjectStorageCommonService::QueryResource(const std::string key, AppProtect::ResourceStatus &ret,
    const std::string& mainJobId, const std::string& subJobId)
{
    int retryCnt = 0;
    Resource res {};
    res.__set_sharedNum(1);
    res.__set_scope(ResourceScope::SINGLE_JOB);
    res.__set_scopeKey(key);
    res.__set_resourceKey(key);
    do {
        if (retryCnt % OBSPlugin::MIDDLE_EXECEPTION_RETRY_TIMES == 0) {
            ObjectStorageCommonService::KeepJobAlive(mainJobId, subJobId);
        }
        try {
            ShareResource::QueryResource(ret, res, mainJobId);
        } catch(AppProtectFrameworkException &e) {
            WARNLOG("QueryResource for key: %s failed, code: %d, retryCnt: %d", key.c_str(), e.code, retryCnt);
            if (e.code == E_QUERIED_RESOURCE_NOT_EXIST) {
                ERRLOG("Exit QueryResource for key: %s, resource not exist!", key.c_str());
                return false;
            }
            Module::SleepFor(std::chrono::seconds(OBSPlugin::EXECEPTION_RETRY_INTERVAL));
            continue;
        }
        DBGLOG("Exit QueryResource for key: %s success", key.c_str());
        return true;
    } while (++retryCnt <= OBSPlugin::EXECEPTION_RETRY_TIMES);
    ERRLOG("Exit QueryResource for key: %s, after max retry & failed", key.c_str());
    return false;
}

bool ObjectStorageCommonService::UpdateResource(const std::string key, Resource &res,
    const std::string& mainJobId, const std::string& subJobId)
{
    int retryCnt = 0;
    ActionResult ret {};
    do {
        if (retryCnt == OBSPlugin::MIDDLE_NORMAL_RETRY_TIMES) {
            ObjectStorageCommonService::KeepJobAlive(mainJobId, subJobId);
        }
        ShareResource::UpdateResource(ret, res, mainJobId);
        if (ret.code != Module::FAILED) {
            DBGLOG("Exit UpdateResource for key: %s success", key.c_str());
            return true;
        }
        WARNLOG("UpdateResource for key: %s failed, retryCnt: %d", key.c_str(), retryCnt);
        Module::SleepFor(std::chrono::seconds(OBSPlugin::NORMAL_RETRY_INTERVAL));
    } while (++retryCnt <= OBSPlugin::NORMAL_RETRY_TIMES);
    ERRLOG("Exit UpdateResource for key: %s, after max retry & failed", key.c_str());
    return false;
}

bool ObjectStorageCommonService::DeleteResource(
    const std::string key, const std::string& mainJobId, const std::string& subJobId)
{
    int retryCnt = 0;
    ActionResult ret {};
    Resource res {};
    res.__set_sharedNum(1);
    res.__set_scope(ResourceScope::SINGLE_JOB);
    res.__set_scopeKey(key);
    res.__set_resourceKey(key);
    do {
        if (retryCnt == OBSPlugin::MIDDLE_NORMAL_RETRY_TIMES) {
            ObjectStorageCommonService::KeepJobAlive(mainJobId, subJobId);
        }
        ShareResource::DeleteResource(ret, res, mainJobId);
        if (ret.code != Module::FAILED) {
            DBGLOG("Exit DeleteResource for key: %s success", key.c_str());
            return true;
        }
        WARNLOG("DeleteResource for key: %s failed, retryCnt: %d, code: %d", key.c_str(), retryCnt, ret.code);
        Module::SleepFor(std::chrono::seconds(OBSPlugin::NORMAL_RETRY_INTERVAL));
    } while (++retryCnt <= OBSPlugin::NORMAL_RETRY_TIMES);
    ERRLOG("Exit DeleteResource for key: %s, after max retry & failed", key.c_str());
    return false;
}

bool ObjectStorageCommonService::LockResource(
    const std::string key, const uint32_t maxRetryCount, const std::string &mainJobId)
{
    uint32_t retryCnt = 0;
    ActionResult ret {};
    ret.code = 0;
    Resource res {};
    res.__set_sharedNum(1);
    res.__set_scope(ResourceScope::SINGLE_JOB);
    res.__set_scopeKey(key);
    res.__set_resourceKey(key);
    INFOLOG("Enter LockResource for key: %s", key.c_str());
    do {
        ShareResource::LockResource(ret, res, mainJobId);
        if (ret.code == MP_SUCCESS) {
            INFOLOG("Exit LockResource, success lock key: %s ", key.c_str());
            return true;
        }
        Module::SleepFor(std::chrono::seconds(NUMBER3));
        WARNLOG("LockResource failed for key: %s, ret.code: %d, ret.bodyErr: %d, ret.message: %s, retryCnt: %d",
            key.c_str(), ret.code, ret.bodyErr, ret.message.c_str(), retryCnt);
    } while (++retryCnt < maxRetryCount);
    ERRLOG("Exit LockResource, failed lock key: %s after max retry, ret.code: %d, ret.message: %s, ret.bodyErr: %d",
        key.c_str(), ret.code, ret.message.c_str(), ret.bodyErr);
    return false;
}

bool ObjectStorageCommonService::UnlockResource(const std::string key, const uint32_t maxRetryCount,
    const std::string& mainJobId)
{
    uint32_t retryCnt = 0;
    ActionResult ret {};
    Resource res {};
    res.__set_sharedNum(1);
    res.__set_scope(ResourceScope::SINGLE_JOB);
    res.__set_scopeKey(key);
    res.__set_resourceKey(key);

    INFOLOG("Enter UnlockResource for key : %s", key.c_str());
    do {
        ShareResource::UnLockResource(ret, res, mainJobId);
        if (ret.code == MP_SUCCESS) {
            INFOLOG("Exit UnlockResource, success unlock key: %s ", key.c_str());
            return true;
        }
        Module::SleepFor(std::chrono::seconds(NUMBER3));
        WARNLOG("UnlockResource failed for key: %s, ret.code: %d, ret.bodyErr: %d, retryCnt: %d",
            key.c_str(), ret.code, ret.bodyErr, retryCnt);
    } while (++retryCnt < maxRetryCount);
    ERRLOG("Exit UnlockResource, failed unlock key: %s after max retry, ret.code: %d, ret.bodyErr: %d",
        key.c_str(), ret.code, ret.bodyErr);
    return false;
}

bool ObjectStorageCommonService::CreateScanStatsResource(std::string key, ObjectStorageNativeScanStatistics &scanStats)
{
    key = key + OBSPlugin::SHARED_RSC_SCANNER_STATS_KEY;
    HCP_Log(DEBUG, MODULE) << "CreateScanStatsResource key is:" << key  << HCPENDLOG;

    std::string scanStatsString;
    if (!Module::JsonHelper::StructToJsonString(scanStats, scanStatsString)) {
        HCP_Log(ERR, MODULE) << "Create resource failed (convert to JSON failed) for key "
            << key << HCPENDLOG;
        return false;
    }
    if (!CreateResource(key, scanStatsString, m_jobId, m_subJobId)) {
        HCP_Log(ERR, MODULE) << "Create resource failed for key " << key;
        return false;
    }
    return true;
}

bool ObjectStorageCommonService::GetScanStatsResource(std::string key, ObjectStorageNativeScanStatistics &scanStats)
{
    key = key + OBSPlugin::SHARED_RSC_SCANNER_STATS_KEY;
    HCP_Log(DEBUG, MODULE) << "GetScanStatsResource key is:" << key  << HCPENDLOG;

    AppProtect::ResourceStatus res {};
    if (!QueryResource(key, res, m_jobId, m_subJobId)) {
        HCP_Log(ERR, MODULE) << "Get resource failed for key " << key << HCPENDLOG;
        return false;
    }
    if (!Module::JsonHelper::JsonStringToStruct(res.resource.resourceValue, scanStats)) {
        HCP_Log(ERR, MODULE) << "Get resource failed (convert to JSON failed) for key " << key << HCPENDLOG;
        return false;
    }
    return true;
}

bool ObjectStorageCommonService::UpdateScanStatsResource(std::string key, ObjectStorageNativeScanStatistics &scanStats)
{
    key = key + OBSPlugin::SHARED_RSC_SCANNER_STATS_KEY;
    HCP_Log(DEBUG, MODULE) << "UpdateScanStatsResource for key: " << key  << HCPENDLOG;

    Resource res {};
    res.__set_sharedNum(1);
    res.__set_scope(ResourceScope::SINGLE_JOB);
    res.__set_scopeKey(key);
    res.__set_resourceKey(key);

    std::string scanStatsString;
    if (!Module::JsonHelper::StructToJsonString(scanStats, scanStatsString)) {
        HCP_Log(ERR, MODULE) << "Update resource failed (convert to JSON failed) for key: " << key << HCPENDLOG;
        return false;
    }

    res.__set_resourceValue(scanStatsString);
    if (!UpdateResource(key, res, m_jobId, m_subJobId)) {
        HCP_Log(ERR, MODULE) << "UpdateScanStatsResource failed for key: " << key << HCPENDLOG;
        return false;
    }

    return true;
}

bool ObjectStorageCommonService::DeleteScanStatsResource(std::string key)
{
    key = key + OBSPlugin::SHARED_RSC_SCANNER_STATS_KEY;
    HCP_Log(DEBUG, MODULE) << "DeleteScanStatsResource key is: " << key  << HCPENDLOG;

    if (!DeleteResource(key, m_jobId, m_subJobId)) {
        HCP_Log(ERR, MODULE) << "DeleteScanStatsResource failed for key: " << key;
        return false;
    }
    return true;
}

bool ObjectStorageCommonService::LockBackupStatsResource(std::string mainJobId)
{
    std::string key = mainJobId + OBSPlugin::SHARED_RSC_BACKUP_STATS_KEY;
    return LockResource(key, NUMBER0, mainJobId);
}

bool ObjectStorageCommonService::UnlockBackupStatsResource(std::string mainJobId)
{
    std::string key = mainJobId + OBSPlugin::SHARED_RSC_BACKUP_STATS_KEY;
    return UnlockResource(key, NUMBER0, mainJobId);
}

bool ObjectStorageCommonService::CreateBackupStatsResource(std::string key, ObjectStorageNativeBackupStats &backupStats)
{
    key = key + OBSPlugin::SHARED_RSC_BACKUP_STATS_KEY;
    HCP_Log(DEBUG, MODULE) << "CreateBackupStatsResource  key is:" << key  << HCPENDLOG;

    std::string backupStatsString;
    if (!Module::JsonHelper::StructToJsonString(backupStats, backupStatsString)) {
        HCP_Log(ERR, MODULE) << "Create resource failed (convert to JSON failed) for key " << key << HCPENDLOG;
        return false;
    }
    if (!CreateResource(key, backupStatsString, m_jobId, m_subJobId)) {
        HCP_Log(ERR, MODULE) << "Create resource failed for key " << key << HCPENDLOG;
        return false;
    }
    return true;
}

bool ObjectStorageCommonService::GetBackupStatsResource(std::string key, ObjectStorageNativeBackupStats &backupStats)
{
    key = key + OBSPlugin::SHARED_RSC_BACKUP_STATS_KEY;
    HCP_Log(DEBUG, MODULE) << "GetBackupStatsResource  key is:" << key  << HCPENDLOG;

    AppProtect::ResourceStatus res {};
    if (!QueryResource(key, res, m_jobId, m_subJobId)) {
        HCP_Log(ERR, MODULE) << "Get resource failed for key: " << key << HCPENDLOG;
        return false;
    }
    if (!Module::JsonHelper::JsonStringToStruct(res.resource.resourceValue, backupStats)) {
        HCP_Log(ERR, MODULE) << "Get resource failed (convert to JSON failed) for key " << key << HCPENDLOG;
        return false;
    }

    return true;
}

bool ObjectStorageCommonService::UpdateBackupStatsResource(std::string key, ObjectStorageNativeBackupStats &backupStats)
{
    key = key + OBSPlugin::SHARED_RSC_BACKUP_STATS_KEY;
    HCP_Log(DEBUG, MODULE) << "UpdateBackupStatsResource for key: " << key  << HCPENDLOG;

    Resource res {};
    res.__set_sharedNum(1);
    res.__set_scope(ResourceScope::SINGLE_JOB);
    res.__set_scopeKey(key);
    res.__set_resourceKey(key);

    std::string backupStatsString;
    if (!Module::JsonHelper::StructToJsonString(backupStats, backupStatsString)) {
        HCP_Log(ERR, MODULE) << "Update resource failed (convert to JSON failed) for key: " << key << HCPENDLOG;
        return false;
    }

    res.__set_resourceValue(backupStatsString);
    if (!UpdateResource(key, res, m_jobId, m_subJobId)) {
        return false;
    }

    return true;
}

bool ObjectStorageCommonService::DeleteBackupStatsResource(std::string key)
{
    key = key + OBSPlugin::SHARED_RSC_BACKUP_STATS_KEY;
    HCP_Log(DEBUG, MODULE) << "DeleteBackupStatsResource key is:" << key << HCPENDLOG;

    if (!DeleteResource(key, m_jobId, m_subJobId)) {
        HCP_Log(ERR, MODULE) << "Delete resource failed for key: " << key << HCPENDLOG;
        return false;
    }
    return true;
}

bool ObjectStorageCommonService::LockGeneralResource(std::string mainJobId)
{
    std::string key = mainJobId + OBSPlugin::SHARED_RSC_GENEAL_INFO_KEY;
    return LockResource(key, NUMBER100, mainJobId);
}

bool ObjectStorageCommonService::UnlockGeneralResource(std::string mainJobId)
{
    std::string key = mainJobId + OBSPlugin::SHARED_RSC_GENEAL_INFO_KEY;
    return UnlockResource(key, NUMBER10, mainJobId);
}

bool ObjectStorageCommonService::CreateGeneralResource(std::string key, ObjectStorageNativeGeneral &generalInfo)
{
    if (GetGeneralResource(key, generalInfo)) {
        HCP_Log(WARN, MODULE) << "Resource for key " << key << " already exist! "
                        << "generalInfo.m_jobStartTime = " << generalInfo.m_jobStartTime
                        << ", generalInfo.m_backupCopyPhaseStartTime = " << generalInfo.m_backupCopyPhaseStartTime
                        << ", generalInfo.m_backupDelPhaseStartTime = " << generalInfo.m_backupDelPhaseStartTime
                        << HCPENDLOG;
        return true;
    }

    key = key + OBSPlugin::SHARED_RSC_GENEAL_INFO_KEY;

    std::string generalInfoString;
    if (!Module::JsonHelper::StructToJsonString(generalInfo, generalInfoString)) {
        HCP_Log(ERR, MODULE) << "Create resource failed (convert to JSON failed) for key " << key << HCPENDLOG;
        return false;
    }
    if (!CreateResource(key, generalInfoString, m_jobId, m_subJobId)) {
        HCP_Log(ERR, MODULE) << "Create resource failed for key:" << key << HCPENDLOG;
        return false;
    }
    HCP_Log(DEBUG, MODULE) << "CreateGeneralResource key: " << key << ", value: " << generalInfoString << HCPENDLOG;
    return true;
}

bool ObjectStorageCommonService::GetGeneralResource(std::string key, ObjectStorageNativeGeneral &generalInfo)
{
    key = key + OBSPlugin::SHARED_RSC_GENEAL_INFO_KEY;
    HCP_Log(DEBUG, MODULE) << "GetGeneralResource key is:" << key << HCPENDLOG;

    AppProtect::ResourceStatus res {};
    if (!QueryResource(key, res, m_jobId, m_subJobId)) {
        HCP_Log(ERR, MODULE) << "Get resource failed for key " << key << HCPENDLOG;
        return false;
    }
    if (!Module::JsonHelper::JsonStringToStruct(res.resource.resourceValue, generalInfo)) {
        HCP_Log(ERR, MODULE) << "Get resource failed (convert to JSON failed) for key " << key << HCPENDLOG;
        return false;
    }

    HCP_Log(DEBUG, MODULE) << "GetGeneralResource key: " << key << ", val: "<< res.resource.resourceValue << HCPENDLOG;
    return true;
}

bool ObjectStorageCommonService::UpdateGeneralResource(std::string key, ObjectStorageNativeGeneral &generalInfo)
{
    key = key + OBSPlugin::SHARED_RSC_GENEAL_INFO_KEY;
    HCP_Log(DEBUG, MODULE) << "UpdateGeneralResource key is:" << key << HCPENDLOG;

    Resource res {};
    res.__set_sharedNum(1);
    res.__set_scope(ResourceScope::SINGLE_JOB);
    res.__set_scopeKey(key);
    res.__set_resourceKey(key);

    std::string generalInfoString;
    if (!Module::JsonHelper::StructToJsonString(generalInfo, generalInfoString)) {
        HCP_Log(ERR, MODULE) << "Update resource failed (convert to JSON failed) for key " << key << HCPENDLOG;
        return false;
    }

    res.__set_resourceValue(generalInfoString);
    if (!UpdateResource(key, res, m_jobId, m_subJobId)) {
        HCP_Log(ERR, MODULE) << "UpdateGeneralResource failed for key: " << key << HCPENDLOG;
        return false;
    }

    HCP_Log(DEBUG, MODULE) << "UpdateGeneralResource key: " << key << ", value: " << generalInfoString << HCPENDLOG;
    return true;
}

bool ObjectStorageCommonService::DeleteGeneralResource(std::string key)
{
    key = key + OBSPlugin::SHARED_RSC_GENEAL_INFO_KEY;
    HCP_Log(DEBUG, MODULE) << "DeleteGeneralResource key is:" << key << HCPENDLOG;

    if (!DeleteResource(key, m_jobId, m_subJobId)) {
        HCP_Log(ERR, MODULE) << "Delete resource failed for key " << key << HCPENDLOG;
        return false;
    }
    return true;
}

uint64_t ObjectStorageCommonService::GetCurrentTimeFromRemoteServer(std::string path)
{
    std::string tempFileName = CreateUniqueTempFileName(path);
    std::ofstream tempFile {};
    uint32_t retryCnt = 0;

    // 1. Create a temp file (dpa_current-time)
    do {
        tempFile.open(tempFileName.c_str(), std::ios::out | std::ios::binary);
        if (tempFile.is_open()) {
            break;
        }
        tempFile.close();
        Module::SleepFor(std::chrono::seconds(NUMBER1));
    } while (++retryCnt < NUMBER3);

    if (!tempFile.is_open()) {
        char errmsg[NUMBER256];
        strerror_r(errno, errmsg, NUMBER256);
        HCP_Log(ERR, MODULE) << "Failed to open file (after 3 retries): " << WIPE_SENSITIVE(tempFileName)
            << ", ERR: " << errmsg << HCPENDLOG;

        return 0;
    }
    tempFile.close();

    // 2. Get the ctime of the temp file
    struct stat info {};
    if (stat(tempFileName.c_str(), &info) != 0) {
        HCP_Log(ERR, MODULE) << "Stat failed for the temp file in remote OBS: "
        << WIPE_SENSITIVE(tempFileName) << HCPENDLOG;
        return 0;
    }
    time_t createTime = info.st_ctime;

    // 3. Delete the file
    try {
        if (!boost::filesystem::remove(tempFileName)) {
            HCP_Log(ERR, MODULE) << "Failure when removing file: " << WIPE_SENSITIVE(tempFileName) << HCPENDLOG;
            return 0;
        }
    } catch (const boost::filesystem::filesystem_error &e) {
        HCP_Log(ERR, MODULE) << "remove() exeption: " << WIPE_SENSITIVE(e.code().message())
        << ", path: " << WIPE_SENSITIVE(tempFileName);
        return 0;
    }

    // 4. return the ctime of the file
    uint64_t currTimeOfRemoteServer = static_cast<uint64_t>(createTime);
    HCP_Log(DEBUG, MODULE) << "Current Time Of Remote Server: " << currTimeOfRemoteServer;
    return currTimeOfRemoteServer;
}

std::string ObjectStorageCommonService::CreateUniqueTempFileName(std::string path)
{
    string tempFileName {};
    bool isFileExist = false;
    boost::system::error_code ec;

    if (m_idGenerator == nullptr) {
        InitIdGenerator();
    }

    do {
        tempFileName = path + "/" + "dpa_" + std::to_string(m_idGenerator->GenerateId());
        try {
            isFileExist = boost::filesystem::exists(tempFileName, ec);
            HCP_Log(DEBUG, MODULE) << "check file exists End, file name=" << tempFileName << HCPENDLOG;

            if (ec.value() != boost::system::errc::success) {
                HCP_Log(DEBUG, MODULE) << "boost::filesystem::exists() reported failure as: " << ec.message()
                    << ". filename: " << tempFileName << HCPENDLOG;
            }
        } catch (const boost::filesystem::filesystem_error &e) {
            HCP_Log(ERR, MODULE) << "boost::filesystem::exists() exeption: " << e.code().message()
                << " path " << tempFileName << HCPENDLOG;
        }
    } while (isFileExist);

    return tempFileName;
}

bool ObjectStorageCommonService::WriteBackupStatsToFile(
    std::string filePath, ObjectStorageNativeBackupStats &backupStatistics)
{
    std::string jsonFileContent {};
    if (!Module::JsonHelper::StructToJsonString(backupStatistics, jsonFileContent)) {
        HCP_Log(ERR, MODULE) << "Write " << WIPE_SENSITIVE(filePath) << " failed, ERR: Json covert failed" << HCPENDLOG;
        return false;
    }
    HCP_Log(INFO, MODULE) << "Subjob Id: " << m_subJobInfo->subJobId
        << " backup json file content: " << jsonFileContent << HCPENDLOG;

    std::vector<std::string> fileContent {};
    fileContent.emplace_back(jsonFileContent);
    return OBSPlugin::WriteFile(filePath, fileContent);
}

bool ObjectStorageCommonService::ReadBackupStatsFromFile(
    std::string filePath, ObjectStorageNativeBackupStats &backupStatistics)
{
    std::vector<std::string> fileContent {};
    if (!OBSPlugin::ReadFile(filePath, fileContent)) {
        return false;
    }
    if (fileContent.empty()) {
        WARNLOG("SubJobStatsFile content is empty");
        return true;
    }
    std::string jsonFileContent = fileContent.front();
    if (jsonFileContent.empty()) {
        WARNLOG("SubJobStatsFile content is empty");
        return true;
    }
    if (!Module::JsonHelper::JsonStringToStruct(jsonFileContent, backupStatistics)) {
        ERRLOG("Read %s failed, ERR: json convert failed", filePath.c_str());
        return false;
    }
    DBGLOG("SubJobStatsFile content: %s", jsonFileContent.c_str());
    return true;
}

bool ObjectStorageCommonService::CreateSharedResources(std::string jobId, ObjectStorageNativeGeneral &generalInfo,
    ObjectStorageNativeScanStatistics &scanStats, ObjectStorageNativeBackupStats &backupStats)
{
    if (!CreateGeneralResource(jobId, generalInfo)) {
        HCP_Log(ERR, MODULE) << "CreateGeneralResource failed" << HCPENDLOG;
        return false;
    }
    if (!CreateScanStatsResource(jobId, scanStats)) {
        HCP_Log(ERR, MODULE) << "CreateScanStatsResource failed" << HCPENDLOG;
        DeleteGeneralResource(jobId);
        return false;
    }
    if (!CreateBackupStatsResource(jobId, backupStats)) {
        HCP_Log(ERR, MODULE) << "CreateBackupStatsResource failed" << HCPENDLOG;
        DeleteGeneralResource(jobId);
        DeleteScanStatsResource(jobId);
        return false;
    }
    return true;
}

bool ObjectStorageCommonService::GetSharedResources(std::string jobId, ObjectStorageNativeGeneral &generalInfo,
    ObjectStorageNativeScanStatistics &scanStats, ObjectStorageNativeBackupStats &backupStats)
{
    if (!GetGeneralResource(jobId, generalInfo)) {
        HCP_Log(ERR, MODULE) << "GetGeneralResource failed" << HCPENDLOG;
        return false;
    }
    if (!GetScanStatsResource(jobId, scanStats)) {
        HCP_Log(ERR, MODULE) << "GetScanStatsResource failed" << HCPENDLOG;
        return false;
    }
    if (!GetBackupStatsResource(jobId, backupStats)) {
        HCP_Log(ERR, MODULE) << "GetBackupStatsResource failed" << HCPENDLOG;
        return false;
    }
    return true;
}

bool ObjectStorageCommonService::DeleteSharedResources(std::string jobId)
{
    if (!DeleteGeneralResource(jobId)) {
        HCP_Log(ERR, MODULE) << "DeleteGeneralResource failed" << HCPENDLOG;
    }
    if (!DeleteScanStatsResource(jobId)) {
        HCP_Log(ERR, MODULE) << "DeleteScanStatsResource failed" << HCPENDLOG;
    }
    if (!DeleteBackupStatsResource(jobId)) {
        HCP_Log(ERR, MODULE) << "DeleteBackupStatsResource failed" << HCPENDLOG;
    }

    return true;
}

bool ObjectStorageCommonService::UpdateScannerStatistics(const ObjectStorageNativeScanStatistics& scanStats,
    ObjectStorageNativeScanStatistics& scanStatistics)
{
    ScanStatistics stats = m_scanner->GetStatistics();
    scanStatistics.m_scanStatus = scanStats.m_scanStatus;
    scanStatistics.m_scanStartTime = scanStats.m_scanStartTime;
    scanStatistics.m_scanEndTime = scanStats.m_scanEndTime;
    scanStatistics.m_remoteObsScanEndTime = scanStats.m_remoteObsScanEndTime;

    scanStatistics.m_totFiles = stats.mTotFiles;
    scanStatistics.m_totFailedFiles = stats.mTotFailedFiles;
    scanStatistics.m_totFilesToBackup = stats.mTotFilesToBackup;
    scanStatistics.m_totFilesDeleted = stats.mTotFilesDeleted;
    scanStatistics.m_totalControlFiles = stats.mTotalControlFiles;
    scanStatistics.m_totalSize = stats.mTotalSize;
    scanStatistics.m_totalSizeToBackup = stats.mTotalSizeToBackup;

    scanStatistics.m_totFailedDirs = stats.mTotFailedDirs; // 列举失败

    PrintScannerStatistics(scanStatistics);

    return true;
}

ScanStatistics ObjectStorageCommonService::AddScanStatistics(
    const ScanStatistics &stats_A, const ScanStatistics &stats_B)
{
    ScanStatistics scanStatistics;
    scanStatistics.mTotDirs = stats_A.mTotDirs + stats_B.mTotDirs;
    scanStatistics.mTotFiles = stats_A.mTotFiles + stats_B.mTotFiles;
    scanStatistics.mTotalSize = stats_A.mTotalSize + stats_B.mTotalSize;
    scanStatistics.mTotDirsToBackup = stats_A.mTotDirsToBackup + stats_B.mTotDirsToBackup;
    scanStatistics.mTotFilesToBackup = stats_A.mTotFilesToBackup + stats_B.mTotFilesToBackup;
    scanStatistics.mTotFilesDeleted = stats_A.mTotFilesDeleted + stats_B.mTotFilesDeleted;
    scanStatistics.mTotDirsDeleted = stats_A.mTotDirsDeleted + stats_B.mTotDirsDeleted;
    scanStatistics.mTotalSizeToBackup = stats_A.mTotalSizeToBackup + stats_B.mTotalSizeToBackup;
    scanStatistics.mTotalControlFiles = stats_A.mTotalControlFiles + stats_B.mTotalControlFiles;
    scanStatistics.mTotFailedDirs = stats_A.mTotFailedDirs + stats_B.mTotFailedDirs;
    scanStatistics.mTotFailedFiles = stats_A.mTotFailedFiles + stats_B.mTotFailedFiles;
    return scanStatistics;
}

void ObjectStorageCommonService::PrintScannerStatistics(ObjectStorageNativeScanStatistics &nativeScanStats)
{
    std::string scanStartTimeStr = FormatTimeToStr(nativeScanStats.m_scanStartTime);
    std::string remoteObsScanEndTimeStr = FormatTimeToStr(nativeScanStats.m_remoteObsScanEndTime);
    std::string scanEndTimeStr = FormatTimeToStr(nativeScanStats.m_scanEndTime);

    int32_t rmtObsScanDur = double(nativeScanStats.m_remoteObsScanEndTime - nativeScanStats.m_scanStartTime);
    int32_t remoteObsScanSpeed = (rmtObsScanDur == 0) ? nativeScanStats.m_totFiles :
        (nativeScanStats.m_totFiles / rmtObsScanDur);

    int32_t scanDuration = double(nativeScanStats.m_scanEndTime - nativeScanStats.m_scanStartTime);
    int32_t scanSpeed = (scanDuration == 0) ? nativeScanStats.m_totFiles : (nativeScanStats.m_totFiles / scanDuration);

    INFOLOG("Scan status               : %d", nativeScanStats.m_scanStatus);
    INFOLOG("Scan start time           : %s", scanStartTimeStr.c_str());
    INFOLOG("# of files scanned        : %llu", nativeScanStats.m_totFiles);
    INFOLOG("# of files Failed to scan : %llu", nativeScanStats.m_totFailedFiles);
    INFOLOG("Total Size                : %s", FormatCapacity(nativeScanStats.m_totalSize).c_str());
    INFOLOG("# of files to bkup        : %llu", nativeScanStats.m_totFilesToBackup);
    INFOLOG("# of files to del         : %llu", nativeScanStats.m_totFilesDeleted);
    INFOLOG("Size to bkup              : %s", FormatCapacity(nativeScanStats.m_totalSizeToBackup).c_str());
    INFOLOG("# control files           : %llu", nativeScanStats.m_totalControlFiles);

    if (nativeScanStats.m_scanStatus == (int)SCANNER_STATUS::SCAN_COMPLETED) {
        INFOLOG("Remote OBS Scan end time  : %s", remoteObsScanEndTimeStr.c_str());
        INFOLOG("Overall Scan end time     : %s", scanEndTimeStr.c_str());
        INFOLOG("Remote OBS Scan duration  : %d sec", rmtObsScanDur);
        INFOLOG("Remote OBS Scan speed     : %d files/sec", remoteObsScanSpeed);
        INFOLOG("Scan duration             : %d sec", scanDuration);
        INFOLOG("Scan speed                : %d files/sec", scanSpeed);
    }
}

void ObjectStorageCommonService::PrintBackupStatistics(BackupStats &backupStatistics,
    std::string &jobId, BackupPhaseStatus m_backupStatus, std::string &statPrintSubJobId)
{
    std::string backupStartTimeStr = FormatTimeToStr(backupStatistics.startTime);
    INFOLOG("Backup Curr Job %s, subjob Cnt: %llu", jobId.c_str(), m_nodeLevelTaskInfo.GetRunningSubTasksCount());
    INFOLOG("Backup Total Job Executed Cnt : %llu", m_nodeLevelTaskInfo.GetSubTasksCount());
    INFOLOG("Backup Total Job Submitted Cnt: %llu", m_nodeLevelTaskInfo.GetSubTasksSubmitCount());
    INFOLOG("Backup status                 : %d, %s", static_cast<int>(m_backupStatus), statPrintSubJobId.c_str());
    INFOLOG("Backup start time             : %s", backupStartTimeStr.c_str());
    INFOLOG("Backup Speed                  : %s", FormatSpeed(backupStatistics.backupspeed).c_str());
    INFOLOG("# noOfDirToBackup             : %llu", backupStatistics.noOfDirToBackup);
    INFOLOG("# noOfFilesToBackup           : %llu", backupStatistics.noOfFilesToBackup);
    INFOLOG("# noOfBytesToBackup           : %s", FormatCapacity(backupStatistics.noOfBytesToBackup).c_str());
    DBGLOG("# noOfDirToDelete             : %llu", backupStatistics.noOfDirToDelete);
    DBGLOG("# noOfFilesToDelete           : %llu", backupStatistics.noOfFilesToDelete);
    INFOLOG("# noOfDirCopied               : %llu", backupStatistics.noOfDirCopied);
    INFOLOG("# noOfFilesCopied             : %llu", backupStatistics.noOfFilesCopied);
    INFOLOG("# noOfBytesCopied             : %s", FormatCapacity(backupStatistics.noOfBytesCopied).c_str());
    DBGLOG("# noOfDirDeleted              : %llu", backupStatistics.noOfDirDeleted);
    DBGLOG("# noOfFilesDeleted            : %llu", backupStatistics.noOfFilesDeleted);
    INFOLOG("# noOfDirFailed               : %llu", backupStatistics.noOfDirFailed);
    INFOLOG("# noOfFilesFailed             : %llu", backupStatistics.noOfFilesFailed);
    DBGLOG("# noOfSrcRetryCount           : %llu", backupStatistics.noOfSrcRetryCount);
    DBGLOG("# noOfDstRetryCount           : %llu", backupStatistics.noOfDstRetryCount);
    DBGLOG("# Backup noOfFailureRecordsWritten: \t%llu", backupStatistics.noOfFailureRecordsWritten);
}

bool ObjectStorageCommonService::UpdateBackupSubTaskStatistics(
    BackupStats &backupStatistics, std::string &metaFsPath, time_t backupStartTime, time_t backupEndTime)
{
    ObjectStorageNativeBackupStats subBackupJobStatistics {};
    std::string filePath {};
    subBackupJobStatistics.m_noOfDirToBackup    = backupStatistics.noOfDirToBackup;
    subBackupJobStatistics.m_noOfFilesToBackup  = backupStatistics.noOfFilesToBackup;
    subBackupJobStatistics.m_noOfBytesToBackup  = backupStatistics.noOfBytesToBackup;
    subBackupJobStatistics.m_noOfDirToDelete    = backupStatistics.noOfDirToDelete;
    subBackupJobStatistics.m_noOfFilesToDelete  = backupStatistics.noOfFilesToDelete;
    subBackupJobStatistics.m_noOfDirCopied      = backupStatistics.noOfDirCopied;
    subBackupJobStatistics.m_noOfFilesCopied    = backupStatistics.noOfFilesCopied;
    subBackupJobStatistics.m_noOfBytesCopied    = backupStatistics.noOfBytesCopied;
    subBackupJobStatistics.m_noOfDirDeleted     = backupStatistics.noOfDirDeleted;
    subBackupJobStatistics.m_noOfFilesDeleted   = backupStatistics.noOfFilesDeleted;
    subBackupJobStatistics.m_noOfDirFailed      = backupStatistics.noOfDirFailed;
    subBackupJobStatistics.m_noOfFilesFailed    = backupStatistics.noOfFilesFailed;
    subBackupJobStatistics.m_noOfSrcRetryCount  = backupStatistics.noOfSrcRetryCount;
    subBackupJobStatistics.m_noOfDstRetryCount  = backupStatistics.noOfDstRetryCount;
    subBackupJobStatistics.m_noOfFailureRecordsWritten  = backupStatistics.noOfFailureRecordsWritten;
    subBackupJobStatistics.m_backupStartTime = backupStartTime;
    subBackupJobStatistics.m_backupEndTime = backupEndTime;

    filePath = metaFsPath + "/obs-statistics/backup-stats-sub-" + m_subJobInfo->subJobId + ".json";

    WriteBackupStatsToFile(filePath, subBackupJobStatistics);
    return true;
}

void ObjectStorageCommonService::PrintSubJobInfo(std::shared_ptr<SubJob> &m_subJobInfo)
{
    HCP_Log(INFO, MODULE) << "m_subJobInfo->jobId       : " << m_subJobInfo->jobId << HCPENDLOG;
    HCP_Log(INFO, MODULE) << "m_subJobInfo->subJobId    : " << m_subJobInfo->subJobId << HCPENDLOG;
    HCP_Log(INFO, MODULE) << "m_subJobInfo->jobType     : " << m_subJobInfo->jobType << HCPENDLOG;
    HCP_Log(INFO, MODULE) << "m_subJobInfo->jobName     : " << m_subJobInfo->jobName << HCPENDLOG;
    HCP_Log(INFO, MODULE) << "m_subJobInfo->jobPriority : " << m_subJobInfo->jobPriority << HCPENDLOG;
    HCP_Log(INFO, MODULE) << "m_subJobInfo->policy      : " << m_subJobInfo->policy << HCPENDLOG;
    HCP_Log(INFO, MODULE) << "m_subJobInfo->jobInfo     : " << m_subJobInfo->jobInfo << HCPENDLOG;
    return;
}

bool ObjectStorageCommonService::IsUpdateBackupStartTimeRequired(ObjectStorageBackupSubJob& backupSubJob,
    ObjectStorageNativeGeneral &generalInfo)
{
    if (backupSubJob.m_SubTaskType == OBSPlugin::SUBJOB_TYPE_DATACOPY_COPY_PHASE) {
        if (generalInfo.m_backupCopyPhaseStartTime == 0 ||
            (generalInfo.m_restoreCopyIndex != backupSubJob.orderNumberForAggregate)) {
            INFOLOG("need to update general info: %s, %d, %d, %d", backupSubJob.m_ControlFile.c_str(),
                generalInfo.m_backupCopyPhaseStartTime, generalInfo.m_restoreCopyIndex,
                backupSubJob.orderNumberForAggregate);
            return true;
        }
    }

    if (backupSubJob.m_SubTaskType == OBSPlugin::SUBJOB_TYPE_DATACOPY_DELETE_PHASE) {
        if (generalInfo.m_backupDelPhaseStartTime == 0 ||
            (generalInfo.m_restoreDeleteIndex != backupSubJob.orderNumberForAggregate)) {
            INFOLOG("need to update general info: %s, %d, %d, %d", backupSubJob.m_ControlFile.c_str(),
                generalInfo.m_backupDelPhaseStartTime, generalInfo.m_restoreDeleteIndex,
                backupSubJob.orderNumberForAggregate);
            return true;
        }
    }

    return false;
}

bool ObjectStorageCommonService::PrintFinalBackupStats(ObjectStorageNativeBackupStats &backupStatistics)
{
    HCP_Log(INFO, MODULE)
        << "\n\n------------------"
        << "\nBackup Statistics"
        << "\n------------------"
        << "\nNo of Dirs Copied          : " << std::to_string(backupStatistics.m_noOfDirCopied)
        << "\nNo of Files Copied         : " << std::to_string(backupStatistics.m_noOfFilesCopied)
        << "\nNo of Bytes Copied         : " << FormatCapacity(backupStatistics.m_noOfBytesCopied)
        << "\nNo of Dir Deleted          : " << std::to_string(backupStatistics.m_noOfDirDeleted)
        << "\nNo of Files Deleted        : " << std::to_string(backupStatistics.m_noOfFilesDeleted)
        << "\nNo of Dir Failed           : " << std::to_string(backupStatistics.m_noOfDirFailed)
        << "\nNo of Files Failed         : " << std::to_string(backupStatistics.m_noOfFilesFailed)
        << "\nNo of Src Retry Count      : " << std::to_string(backupStatistics.m_noOfSrcRetryCount)
        << "\nNo of Dst Retry Count      : " << std::to_string(backupStatistics.m_noOfDstRetryCount)
        << "\nNo of Failed record Written: " << std::to_string(backupStatistics.m_noOfFailureRecordsWritten)
        << "\n\n" << HCPENDLOG;
    return true;
}

bool ObjectStorageCommonService::PrintFinalScannerStats(ObjectStorageNativeScanStatistics &scanStats)
{
    std::string scanStartTimeStr = FormatTimeToStr(scanStats.m_scanStartTime);
    std::string remoteObsScanEndTimeStr = FormatTimeToStr(scanStats.m_remoteObsScanEndTime);
    std::string scanEndTimeStr = FormatTimeToStr(scanStats.m_scanEndTime);

    int32_t rmtObsScanDur = double(scanStats.m_remoteObsScanEndTime - scanStats.m_scanStartTime);
    int32_t remoteObsScanSpeed = (rmtObsScanDur == 0) ? scanStats.m_totFiles : (scanStats.m_totFiles/rmtObsScanDur);

    int32_t scanDuration = double(scanStats.m_scanEndTime - scanStats.m_scanStartTime);
    int32_t scanSpeed = (scanDuration == 0) ? scanStats.m_totFiles : (scanStats.m_totFiles / scanDuration);

    if (scanStats.m_scanStatus == (int)SCANNER_STATUS::SCAN_COMPLETED) {
        HCP_Log(INFO, MODULE) << "\n\n------------------" << "\nScanner Statistics" << "\n------------------"
            << "\nScan status                : " << "Completed"
            << "\nScan start time            : " << scanStartTimeStr
            << "\nRemote OBS Scan end time   : " << remoteObsScanEndTimeStr
            << "\nRemote OBS Scan duration   : " << std::to_string(rmtObsScanDur) << " sec"
            << "\nRemote OBS Scan speed      : " << std::to_string(remoteObsScanSpeed) << " files/sec"
            << "\nScan duration              : " << std::to_string(scanDuration) << " sec"
            << "\nScan speed                 : " << std::to_string(scanSpeed) << " files/sec"
            << "\nNo of dirs scanned         : " << std::to_string(scanStats.m_totDirs)
            << "\nNo of files scanned        : " << std::to_string(scanStats.m_totFiles)
            << "\nNo of dirs Failed to scan  : " << std::to_string(scanStats.m_totFailedDirs)
            << "\nNo of files Failed to scan : " << std::to_string(scanStats.m_totFailedFiles)
            << "\nTotal Size                 : " << FormatCapacity(scanStats.m_totalSize)
            << "\nNo of dirs to bkup         : " << std::to_string(scanStats.m_totDirsToBackup)
            << "\nNo of files to bkup        : " << std::to_string(scanStats.m_totFilesToBackup)
            << "\nNo of dirs to del          : " << std::to_string(scanStats.m_totDirsDeleted)
            << "\nNo of files to del         : " << std::to_string(scanStats.m_totFilesDeleted)
            << "\nSize to bkup               : " << FormatCapacity(scanStats.m_totalSizeToBackup)
            << "\nNo of control files        : " << std::to_string(scanStats.m_totalControlFiles)
            << "\n\n" << HCPENDLOG;
    } else {
        HCP_Log(INFO, MODULE) << "\n\n------------------" << "\nScanner Statistics" << "\n------------------"
            << "\nScan status               : " << scanStats.m_scanStatus
            << "\nScan start time           : " << scanStartTimeStr
            << "\n# of dirs scanned         : " << std::to_string(scanStats.m_totDirs)
            << "\n# of files scanned        : " << std::to_string(scanStats.m_totFiles)
            << "\n# of dirs Failed to scan  : " << std::to_string(scanStats.m_totFailedDirs)
            << "\n# of files Failed to scan : " << std::to_string(scanStats.m_totFailedFiles)
            << "\nTotal Size                : " << FormatCapacity(scanStats.m_totalSize)
            << "\n# of dirs to bkup         : " << std::to_string(scanStats.m_totDirsToBackup)
            << "\n# of files to bkup        : " << std::to_string(scanStats.m_totFilesToBackup)
            << "\n# of dirs to del          : " << std::to_string(scanStats.m_totDirsDeleted)
            << "\n# of files to del         : " << std::to_string(scanStats.m_totFilesDeleted)
            << "\nSize to bkup              : " << FormatCapacity(scanStats.m_totalSizeToBackup)
            << "\n# control files           : " << std::to_string(scanStats.m_totalControlFiles)
            << "\n\n" << HCPENDLOG;
    }

    return true;
}

void ObjectStorageCommonService::PrintBackupScanFilter(ScanDirectoryFilter dirFilter, ScanFileFilter fileFilter)
{
    HCP_Log(INFO, MODULE) << "fileFilter.type:"  << fileFilter.type << HCPENDLOG;
    HCP_Log(INFO, MODULE) << "fileFilter.fileList:"   << HCPENDLOG;
    for (uint16_t i = 0; i < fileFilter.fileList.size(); i++) {
        HCP_Log(INFO, MODULE) << fileFilter.fileList[i] << HCPENDLOG;
    }
    HCP_Log(INFO, MODULE) << "dirFilter.type:" << dirFilter.type << HCPENDLOG;
    HCP_Log(INFO, MODULE) << "scanFilter.directoryFilter.filterList:" << HCPENDLOG;
    for (uint16_t i = 0; i < dirFilter.dirList.size(); i++) {
        HCP_Log(INFO, MODULE) << dirFilter.dirList[i] << HCPENDLOG;
    }
}

void ObjectStorageCommonService::SetJobCtrlPhase(std::string jobCtrlPhase)
{
    m_jobCtrlPhase = jobCtrlPhase;
    HCP_Log(INFO, MODULE) << "Set job phase - "  << jobCtrlPhase << HCPENDLOG;
}

void ObjectStorageCommonService::SetMainJobId(const string& jobId)
{
    m_jobId = jobId;
}

void ObjectStorageCommonService::SetSubJobId()
{
    m_subJobId = GetSubJobId();
}

void ObjectStorageCommonService::FillScanConfigForNative(ScanConfig &scanConfig)
{
    scanConfig.scanCopyCtrlFileSize = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
        "DME_OBS_SCAN_CTRL_FILE_SIZE");
    scanConfig.scanCtrlMaxDataSize = GetConfigStrKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
        "DME_OBS_SCAN_CTRL_MAX_DATASIZE");
    scanConfig.scanCtrlMinDataSize = GetConfigStrKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
        "DME_OBS_SCAN_CTRL_MIN_DATASIZE");
    scanConfig.scanCtrlMaxEntriesFullBkup = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
        "DME_OBS_SCAN_CTRL_MAX_ENTRIES_FULLBKUP");
    scanConfig.scanCtrlMaxEntriesIncBkup = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
        "DME_OBS_SCAN_CTRL_MAX_ENTRIES_INCRBKUP");
    scanConfig.scanCtrlMinEntriesFullBkup = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
        "DME_OBS_SCAN_CTRL_MIN_ENTRIES_FULLBKUP");
    scanConfig.scanCtrlMinEntriesIncBkup = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
        "DME_OBS_SCAN_CTRL_MIN_ENTRIES_INCRBKUP");
    scanConfig.scanCtrlFileTimeSec = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
        "DME_OBS_SCAN_CTRL_FILE_TIME_SEC");
    scanConfig.obs.onceListNum = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
        "DME_OBS_SCAN_LIST_OBJECTS_NUM");
}

void ObjectStorageCommonService::KeepJobAlive(const string& mainJobId, const string& subJobId)
{
    if (mainJobId.empty()) {
        WARNLOG("mainJobId is empty, can not do KeepJobAlive");
        return;
    }
    SubJobDetails subJobDetails {};
    ActionResult result {};
    subJobDetails.__set_jobId(mainJobId);
    if (!subJobId.empty()) {
        subJobDetails.__set_subJobId(subJobId);
    }
    subJobDetails.__set_jobStatus(SubJobStatus::RUNNING);
    INFOLOG("Enter ReportJobDetails: jobId: %s, subJobId: %s, jobStatus: RUNNING", mainJobId.c_str(), subJobId.c_str());
    uint32_t retryCnt = 0;
    do {
        JobService::ReportJobDetails(result, subJobDetails);
        if (result.code == Module::SUCCESS) {
            HCP_Log(DEBUG, MODULE) << "Exit ReportJobDetails: "
                << "jobId: " << mainJobId
                << ", subJobId: " << subJobId
                << ", jobStatus: RUNNING" << HCPENDLOG;
            return;
        }
        WARNLOG("keep job alive failed, retryCnt: %d, code: %d, jobId: %s", retryCnt, result.code, mainJobId.c_str());
        Module::SleepFor(std::chrono::seconds(NUMBER4));
    } while (++retryCnt <= NUMBER5);
    ERRLOG("keep job alive failed after max retry, jobId: %s, subJobId: %s", mainJobId.c_str(), subJobId.c_str());
}

bool ObjectStorageCommonService::SendJobReportForAliveness()
{
    int64_t currTime = GetCurrentTimeInSeconds();
    if ((currTime - m_lastJobReportTime) > OBSPlugin::REPORT_INTERVAL) {
        KeepJobAlive(m_jobId, m_subJobId);
        m_lastJobReportTime = GetCurrentTimeInSeconds();
    }
    return true;
}

void ObjectStorageCommonService::SleepForCreateSubTaskError()
{
    uint32_t count = 0;
    do {
        if (IsAbortJob()) {
            WARNLOG("Exit received Abort for taskid: %s, subtaskid: %s", m_jobId.c_str(), m_subJobId.c_str());
            break;
        }
        Module::SleepFor(std::chrono::seconds(NUMBER60));
        SendJobReportForAliveness();
    } while (count++ < NUMBER3);
}

bool ObjectStorageCommonService::CheckAndRetryCreateSubTask(std::vector<SubJob> &subJobList,
    std::vector<std::string> &ctrlFileList, int64_t &createJobErrTime, bool isFinal, bool isCheckPoint)
{
    if (subJobList.empty() || ctrlFileList.empty()) {
        DBGLOG("subJobList or ctrlFileList are empty, exit CreateSubTask");
        return true;
    }
    DBGLOG("Enter with subJobList size = %zu, ctrlFileList.size()=%zu", subJobList.size(), ctrlFileList.size());
    do {
        if (IsAbortJob()) {
            WARNLOG("Exit received Abort for taskid: %s, subtaskid: %s", m_jobId.c_str(), m_subJobId.c_str());
            break;
        }
        int ret = CreateSubTask(subJobList, ctrlFileList, isCheckPoint);
        if (ret == Module::SUCCESS) {
            HCP_Log(DEBUG, MODULE) << "Success to create subJobList." << HCPENDLOG;
            break;
        }
        if (ret == Module::RETRY) {
            HCP_Log(WARN, MODULE) << "Create subtask failed with retriable error"<< HCPENDLOG;
            createJobErrTime = GetCurrentTimeInSeconds();
            // is final false means still scanning , true means scan finish.
            // if still scanning, can return here, if scan finish , has to create subtask here, can not return.
            if (!isFinal) {
                subJobList.clear();
                ctrlFileList.clear();
                return true;
            } else {
                HCP_Log(WARN, MODULE) << "sleep for create subtask error"<< HCPENDLOG;
                SleepForCreateSubTaskError();
                continue;
            }
        } else {
            HCP_Log(ERR, MODULE) << "Exit CreateSubTasksFromCtrlFile, Create subtask failed" << HCPENDLOG;
            return false;
        }
    } while (true);
    return true;
}

void ObjectStorageCommonService::MergeBackupFailureRecords()
{
    try {
        INFOLOG("start to merge failure records, output directory root: path %s, jobID: %s",
            m_failureRecordRoot.c_str(), m_jobId.c_str());
        if (!boost::filesystem::exists(m_failureRecordRoot) || !boost::filesystem::is_directory(m_failureRecordRoot)) {
            WARNLOG("failed to merge failure records, dir %s not exists", m_failureRecordRoot.c_str());
            return;
        }
        if (!Module::BackupFailureRecorder::Merge(m_failureRecordRoot, m_jobId, m_failureRecordRoot)) {
            WARNLOG("merge backup failure records failed");
        }
    } catch (const std::exception& e) {
        ERRLOG("merge backup failure records failed with exception: %s", e.what());
    }
}

std::string ObjectStorageCommonService::GetScanGenFileRelativeDir(const PrefixInfo& prefixInfo)
{
    std::string prefixDirName = GetScanPrefixHash(prefixInfo);
    return dir_sep + prefixInfo.bucketName + dir_sep + prefixDirName + dir_sep;
}

std::string ObjectStorageCommonService::GetScanPrefixHash(const PrefixInfo& prefixInfo)
{
    std::string prefixDirName;
    if (prefixInfo.subPrefixs.empty()) {
        prefixDirName = "0";
    } else {
        std::vector<std::string> subPrefixs(prefixInfo.subPrefixs);
        sort(subPrefixs.begin(), subPrefixs.end());
        std::hash<std::string> strHash;
        prefixDirName = std::to_string(strHash(subPrefixs.front()));
    }
    return prefixDirName;
}
bool ObjectStorageCommonService::UpdateScannerSubTaskStatistics(const PrefixInfo& prefixInfo,
    ObjectStorageNativeScanStatistics& scanStatistics,
    const std::string &cacheFsPath, const std::string& subJobId)
{
    std::string fileName = subJobId + prefixInfo.bucketName + GetScanPrefixHash(prefixInfo);
    std::string filePath = cacheFsPath + "/scan-stats-sub-" + fileName + ".json";
    if (!JsonFileTool::WriteToFile(scanStatistics, filePath)) {
        ERRLOG("write scanStatistics failed");
        return false;
    }
    return true;
}

bool ObjectStorageCommonService::GetBackupScanStats(
    const std::string& scanStatPath, ObjectStorageNativeScanStatistics& scanStatistics)
{
    std::vector<std::string> fileList {};
    if (!GetFileListInDirectory(scanStatPath, fileList)) {
        HCP_Log(ERR, MODULE) << "Get filelist for dir failed: " << scanStatPath << HCPENDLOG;
        return false;
    }

    bool isMainStatExist = false;
    for (std::size_t i = 0; i < fileList.size(); i++) {
        if (fileList[i].find("scan-stats-main-") == std::string::npos) {
            continue;
        }

        isMainStatExist = true;
        HCP_Log(INFO, MODULE) << "Get scan-stats file " << fileList[i] << HCPENDLOG;
        ObjectStorageNativeScanStatistics copyScanStatistics {};
        if (!JsonFileTool::ReadFromFile(fileList[i], copyScanStatistics)) {
            HCP_Log(ERR, MODULE) << "Read copy scanfile failed: " << fileList[i] << HCPENDLOG;
            return false;
        }

        scanStatistics.m_totFiles += copyScanStatistics.m_totFiles;
        scanStatistics.m_totalSize += copyScanStatistics.m_totalSize;
        scanStatistics.m_totFailedFiles += copyScanStatistics.m_totFailedFiles;
        scanStatistics.m_totFilesToBackup += copyScanStatistics.m_totFilesToBackup;
        scanStatistics.m_totalSizeToBackup += copyScanStatistics.m_totalSizeToBackup;
    }

    if (!isMainStatExist) {
        HCP_Log(ERR, MODULE) << "Cannot backup stats from file." << HCPENDLOG;
        return false;
    }

    return true;
}

bool ObjectStorageCommonService::CalcSubScanStats(const std::string& scanStatPath,
    bool isTearDown, ObjectStorageNativeScanStatistics& scanStatistics)
{
    std::vector<std::string> fileList {};
    if (!GetFileListInDirectory(scanStatPath, fileList)) {
        HCP_Log(ERR, MODULE) << "Get filelist for dir failed: " << scanStatPath << HCPENDLOG;
        return false;
    }
    for (std::size_t i = 0; i < fileList.size(); i++) {
        if (fileList[i].find("scan-stats-sub") == std::string::npos) {
            HCP_Log(INFO, MODULE) << "File " << fileList[i] << " not need to calc" << HCPENDLOG;
            continue;
        }
        ObjectStorageNativeScanStatistics subScanJobStatistics {};
        if (!JsonFileTool::ReadFromFile(fileList[i], subScanJobStatistics)) {
            continue;
        }
        scanStatistics.m_totFiles += subScanJobStatistics.m_totFiles;
        scanStatistics.m_totalSize += subScanJobStatistics.m_totalSize;
        scanStatistics.m_totFailedFiles += subScanJobStatistics.m_totFailedFiles;
        scanStatistics.m_totFilesToBackup += subScanJobStatistics.m_totFilesToBackup;
        scanStatistics.m_totalSizeToBackup += subScanJobStatistics.m_totalSizeToBackup;
        if (i % NUMBER50 == 0) {
            ReportJobDetails(JobLogLevel::TASK_LOG_INFO, SubJobStatus::RUNNING, OBSPlugin::PROGRESS0);
        }
        if (isTearDown) {
            OBSPlugin::RemoveFile(fileList[i]);
        }
    }
    return true;
}

void ObjectStorageCommonService::RemoveScanStatFile(const PrefixInfo& prefixInfo, const std::string& subJobId,
    const std::string &cacheFsPath)
{
    std::string fileName = subJobId + prefixInfo.bucketName + GetScanPrefixHash(prefixInfo);
    std::string filePath = cacheFsPath + "/scan-stats-sub-" + fileName + ".json";
    OBSPlugin::RemoveFile(filePath);
}

bool ObjectStorageCommonService::LockScanStatsResource(const std::string& mainJobId)
{
    std::string key = mainJobId + OBSPlugin::SHARED_RSC_SCANNER_STATS_KEY;
    DBGLOG("LockScanStatsResource key is:%s", key.c_str());
    return LockResource(key, NUMBER0, mainJobId);
}

bool ObjectStorageCommonService::UnlockScanStatsResource(const std::string& mainJobId)
{
    std::string key = mainJobId + OBSPlugin::SHARED_RSC_SCANNER_STATS_KEY;
    DBGLOG("UnlockScanStatsResource key is:%s", key.c_str());
    return UnlockResource(key, NUMBER0, mainJobId);
}

bool ObjectStorageCommonService::UpdateScanStatsResourceWithLock(ObjectStorageNativeScanStatistics& scanStatistics,
    bool& bReportToPM)
{
    // 上报统计数据给PM
    if (!LockScanStatsResource(m_jobId)) {
        // plugin restart, UBC deadlock, try to unlock
        UnlockScanStatsResource(m_jobId);
        ReportJobDetails(JobLogLevel::TASK_LOG_INFO, SubJobStatus::RUNNING, OBSPlugin::PROGRESS0);
        return false;
    }
    bReportToPM = false;
    ObjectStorageNativeScanStatistics scanStatsResource {};
    if (!GetScanStatsResource(m_jobId, scanStatsResource)) {
        ReportJobDetails(JobLogLevel::TASK_LOG_INFO, SubJobStatus::RUNNING, OBSPlugin::PROGRESS0);
        UnlockScanStatsResource(m_jobId);
        return false;
    }

    if ((GetCurrentTimeInSeconds() - scanStatsResource.m_lastLogReportTime) > SCANNER_REPORT_CIRCLE_TIME) {
        bReportToPM = true;
        scanStatsResource.m_lastLogReportTime = GetCurrentTimeInSeconds();
    }
    scanStatistics.m_lastLogReportTime = scanStatsResource.m_lastLogReportTime;
    if (bReportToPM) {
        UpdateScanStatsResource(m_jobId, scanStatistics);
    }
    UnlockScanStatsResource(m_jobId);
    return true;
}
bool ObjectStorageCommonService::CreateSubTaskWithRetry(uint32_t subTaskType,
    const std::string& jobName, uint32_t jobPriority, bool ignoreFailed)
{
    ObjectStorageBackupSubJob backupSubJob {};
    backupSubJob.m_SubTaskType = subTaskType;
    std::string backupSubJobStr {};
    if (!Module::JsonHelper::StructToJsonString(backupSubJob, backupSubJobStr)) {
        HCP_Log(ERR, MODULE) << "Exit CreateBackupJobTeardownTask failed" << HCPENDLOG;
        return false;
    }

    SubJob subJob {};
    subJob.__set_jobId(m_jobId);
    subJob.__set_jobName(jobName);
    subJob.__set_jobType(SubJobType::BUSINESS_SUB_JOB);
    subJob.__set_policy(ExecutePolicy::ANY_NODE);
    subJob.__set_jobInfo(backupSubJobStr);
    subJob.__set_jobPriority(jobPriority);
    subJob.__set_ignoreFailed(ignoreFailed);

    do {
        if (IsAbortJob()) {
            HCP_Log(WARN, MODULE) << "Exit received Abort for taskid: " << m_jobId << ", subtaskid: "
                << m_subJobId << HCPENDLOG;
            break;
        }
        int ret = CreateSubTask(subJob);
        if (ret == Module::SUCCESS) {
            break;
        } else if (ret == Module::RETRY) {
            HCP_Log(WARN, MODULE) << "Create subtask failed with retriable error"<< HCPENDLOG;
            SleepForCreateSubTaskError();
            continue;
        } else {
            HCP_Log(ERR, MODULE) << "Exit CreateBackupJobTeardownTask failed,subTaskType:" << subTaskType << HCPENDLOG;
            return false;
        }
    } while (true);
    return true;
}

bool ObjectStorageCommonService::CheckEncode(const std::string &endPoint, const std::string &bucketName,
    std::unordered_map<std::string, std::vector<std::string>>& bucketPrefixMap, bool &encodeSwitch)
{
    DBGLOG("CheckEncode, endPoint:%s, bucketName:%s", endPoint.c_str(), bucketName.c_str());
    if (!encodeSwitch) {
        return false;
    }

    Module::StorageType storageType = m_authInfo.getStorageType();
    if (storageType != Module::StorageType::HUAWEI) {
        return false;
    }

    std::string ip = endPoint;
    std::size_t pos = endPoint.find(':'); // 去除端口
    if (pos != std::string::npos) {
        ip = endPoint.substr(0, pos);
    }

    auto it = bucketPrefixMap.find(ip);
    if (it != bucketPrefixMap.end()) {
        const std::vector<std::string>& bucketList = it->second;
        return std::find(bucketList.begin(), bucketList.end(), bucketName) != bucketList.end();
    }
    return false;
}

void ObjectStorageCommonService::ReadEncodeBucket(std::unordered_map<std::string, std::vector<std::string>>& bucketPrefixMap,
    bool &encodeSwitch)
{
    if (m_authInfo.getStorageType() != Module::StorageType::HUAWEI) {
        return;
    }
    
    int encodeFlag = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION, "DME_OBS_ENCODING_FLAG");
    if (encodeFlag != 1) {
        DBGLOG("encodeSwitch is off %d", encodeFlag);
        return;
    }
    
    std::string bucketList = GetConfigStrKey(DME_OBS_CONFIG_SECTION, "DME_OBS_ENCODING_CFG");
    std::istringstream ipStream(bucketList);
    std::string bucketEntry;

    while (std::getline(ipStream, bucketEntry, ';')) {
        std::size_t pos = bucketEntry.find(':');
        if (pos != std::string::npos) {
            std::string ip = bucketEntry.substr(0, pos);
            std::string buckets = bucketEntry.substr(pos + 1);
            std::istringstream bucketStream(buckets);
            std::string bucket;

            while (std::getline(bucketStream, bucket, ',')) {
                bucketPrefixMap[ip].push_back(bucket);
                encodeSwitch = true;
            }
        }
    }
    if (!encodeSwitch) {
        ERRLOG("check encode format");
    }
}

bool ObjectStorageCommonService::CreateZeroBackupJudgeTask(uint32_t jobPriority)
{
    // 用于控制任务在部分成功任务备份或恢复零个数据时，将部分成功状态转变为失败
    std::string jobName = SUBJOB_TYPE_ZERO_BACKUP_JUDGE_JOBNAME;
    return CreateSubTaskWithRetry(SUBJOB_TYPE_ZERO_BACKUP_JUDGE_PHASE, jobName, jobPriority, false);
}

int ObjectStorageCommonService::ExecuteZeroBackupJudgeSubJob(const std::string& backupStatsFilePath)
{
    ObjectStorageNativeBackupStats backupStatistics {};
    ReadBackupStatsFromFile(backupStatsFilePath, backupStatistics);
    if (backupStatistics.m_noOfFilesFailed != 0 && backupStatistics.m_noOfFilesCopied == 0) {
        HCP_Log(WARN, MODULE) << "backup partially succeeded with no object, set task failed" << HCPENDLOG;
        ReportJobDetails(JobLogLevel::TASK_LOG_ERROR, SubJobStatus::FAILED, OBSPlugin::PROGRESS0);
    } else {
        ReportJobDetails(JobLogLevel::TASK_LOG_INFO, SubJobStatus::COMPLETED, OBSPlugin::PROGRESS100);
    }
    return Module::SUCCESS;
}

void ObjectStorageCommonService::KeepSubJobAlive(bool& keepAlive)
{
    HCP_Log(INFO, MODULE) << "Start keep alive for task " << m_jobId << " sub task " << m_subJobId << HCPENDLOG;
    while (keepAlive) {
        ReportJobDetailsWithLabel(SubJobStatus::RUNNING, OBSPlugin::PROGRESS0, "", JobLogLevel::TASK_LOG_INFO);
        std::this_thread::sleep_for(std::chrono::seconds(EXECUTE_SUBTASK_MONITOR_DUR_IN_SEC));
    }
    ReportJobDetails(JobLogLevel::TASK_LOG_INFO, SubJobStatus::COMPLETED, OBSPlugin::PROGRESS100);
    HCP_Log(INFO, MODULE) << "End keep alive for task " << m_jobId << " sub task " << m_subJobId << HCPENDLOG;
}

int ObjectStorageCommonService::CopyUnCompressMeta(const std::string& srcDir, const std::string& targetDir)
{
    HCP_Log(INFO, MODULE) << "Enter copy uncompress meta data." << HCPENDLOG;
    std::string moveToDirCmd = "cd " + srcDir + ";";
    std::string copyMetaCmd = "find . -name '*' -exec cp -r '{}' " + targetDir +  " \\;";
    std::string execCmd = moveToDirCmd + copyMetaCmd;
    HCP_Log(INFO, MODULE) << "run cmd : " << execCmd << HCPENDLOG;
    std::vector<std::string> paramList {};
    int ret = OBSPlugin::RunShellCmd(execCmd, paramList);
    if (ret != 0) {
        HCP_Log(ERR, MODULE) << "Copy meta files failed." << HCPENDLOG;
        return ret;
    }
    HCP_Log(INFO, MODULE) << "Copy meta files finish." << HCPENDLOG;
    return Module::SUCCESS;
}

int ObjectStorageCommonService::GetConfigIntKey(const std::string & sectionName, const std::string & keyName)
{
    if (IsDirExist(OBS_CONFIG_FILE)) {
        return m_objectConfigManager.GetInnerConfigInt(keyName);
    } else {
        return Module::ConfigReader::getInt(sectionName, keyName);
    }
}

std::string ObjectStorageCommonService::GetConfigStrKey(const std::string & sectionName, const string & keyName)
{
    if (IsDirExist(OBS_CONFIG_FILE)) {
        return m_objectConfigManager.GetInnerConfigStr(keyName);
    } else {
        return Module::ConfigReader::getString(sectionName, keyName);
    }
}
}